package cn.ddiancan.schedule.base.service.impl;

import java.net.InetAddress;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.ZoneId;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;

import cn.ddiancan.schedule.base.dao.ITaskTimerDao;
import cn.ddiancan.schedule.base.domain.CronTypeEum;
import cn.ddiancan.schedule.base.domain.ExecuteTypeEnum;
import cn.ddiancan.schedule.base.domain.OperationType;
import cn.ddiancan.schedule.base.domain.TaskDTO;
import cn.ddiancan.schedule.base.domain.TaskLogVO;
import cn.ddiancan.schedule.base.service.ITaskTimerService;
import cn.ddiancan.schedule.business.service.IScheduleExecutor;
import cn.ddiancan.schedule.task.CronTaskRegistrar;
import cn.ddiancan.schedule.task.service.DistributedLockService;
import cn.ddiancan.schedule.task.service.SchedulerService;
import cn.ddiancan.schedule.task.service.SchedulingRunnable;
import cn.ddiancan.xddcloud.common.constants.XddHeaderConstant;
import cn.ddiancan.xddcloud.common.context.RequestContext;
import cn.ddiancan.xddcloud.common.entity.CommonResult;
import cn.ddiancan.xddcloud.common.entity.UserVO;
import cn.ddiancan.xddcloud.ddccore.request.PageRequest;
import cn.ddiancan.xddcloud.ddccore.response.PagedResultListInfo;
import cn.ddiancan.xddcloud.psc.base.service.UserClientService;
import com.alibaba.cloud.nacos.NacosServiceManager;
import com.alibaba.fastjson.JSON;
import com.alibaba.nacos.api.naming.NamingService;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.github.pagehelper.PageHelper;
import lombok.SneakyThrows;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration;
import org.springframework.boot.autoconfigure.web.ServerProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Lazy;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.config.CronTask;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import org.springframework.web.client.RestTemplate;

import static cn.ddiancan.schedule.task.service.SchedulerService.extutorMap;

/**
 * 定时任务基础数据服务类
 *
 * @author Amos
 */
@Service
public class TaskTimerService implements ITaskTimerService {

    private static final Logger logger = LoggerFactory.getLogger(TaskTimerService.class);

    private static final String SCHEDULE_PREFIX = "SCHEDULE_";

    @Autowired
    private ITaskTimerDao taskTimerDao;

    @Autowired
    private CronTaskRegistrar cronTaskRegistrar;

    @Autowired
    private ApplicationContext applicationContext;

    @Autowired
    @Qualifier(value = TaskExecutionAutoConfiguration.APPLICATION_TASK_EXECUTOR_BEAN_NAME)
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Autowired
    private NacosServiceManager nacosServiceManager;

    @Autowired
    private ServerProperties serverProperties;

    @Value("${spring.application.name}")
    private String applicationName;

    @Autowired
    private RestTemplate restTemplate;

    @Autowired
    private DistributedLockService distributedLockService;

    @Autowired
    @Lazy
    private UserClientService userClientService;


    /**
     * 新增任务
     *
     * @param task
     * @return
     */
    @Override
    public CommonResult<Integer> addTaskTimer(TaskDTO task) {
        checkValidDate(task);
        CommonResult<Integer> commonResult = new CommonResult<>();
        commonResult.setCode(HttpStatus.OK.value());
        convertDataToExpression(task);
        task.setScheduleCode(getScheduleCode());
        UserVO currentContextUser = RequestContext.getCurrentContextUser();
        task.setCreatedBy(currentContextUser.getUserId());
        task.setLastUpdateBy(currentContextUser.getUserId());
        commonResult.setData(taskTimerDao.addTaskTimer(task));
        // 更新任务执行清单列表,新增启用状态
        flushScheduleTask(task, OperationType.INSERT);
        // 同步其他片区
        syncPeerNode(task, OperationType.INSERT.name());
        return commonResult;
    }

    public CommonResult<String> flushScheduleTask(TaskDTO task, String operationType) {
        flushScheduleTask(task, OperationType.valueOf(operationType));
        return CommonResult.success();
    }

    private void flushScheduleTask(TaskDTO task, OperationType operationType) {
        switch (operationType) {
            case INSERT -> {
                if ("1".equals(task.getState())) {
                    CronTask cronTask =
                        new CronTask(new SchedulingRunnable(task, applicationContext, distributedLockService),
                            task.getCron());
                    extutorMap.put(task.getScheduleCode(), cronTask);
                    cronTaskRegistrar.addCronTask(cronTask);
                    logger.info("新增加任务：{},新表达式：{}", task.getScheduleName(), cronTask.getExpression());
                }
            }
            case UPDATE -> {
                if (extutorMap.containsKey(task.getScheduleCode())) {
                    CronTask cronTask = extutorMap.get(task.getScheduleCode());
                    SchedulingRunnable runnable = (SchedulingRunnable) cronTask.getRunnable();
                    if ("1".equals(task.getState())) {
                        CronTask newCronTask =
                            new CronTask(new SchedulingRunnable(task, applicationContext, distributedLockService),
                                task.getCron());
                        cronTaskRegistrar.removeCronTask(runnable);
                        extutorMap.remove(task.getScheduleCode());
                        extutorMap.put(task.getScheduleCode(), newCronTask);
                        cronTaskRegistrar.addCronTask(newCronTask);
                        logger.info("已更新任务：{},新表达式：{}", task.getScheduleName(), newCronTask.getExpression());
                    } else {
                        // 失效任务
                        extutorMap.remove(task.getScheduleCode());
                        cronTaskRegistrar.removeCronTask(runnable);
                    }
                }
            }
            case DELETE -> {
                if (extutorMap.containsKey(task.getScheduleCode())) {
                    CronTask cronTask = extutorMap.get(task.getScheduleCode());
                    SchedulingRunnable runnable = (SchedulingRunnable) cronTask.getRunnable();
                    cronTaskRegistrar.removeCronTask(runnable);
                    extutorMap.remove(task.getScheduleCode());
                }
            }
            default -> throw new RuntimeException("操作类型错误");
        }
    }

    private void syncPeerNode(TaskDTO task, String operationType) {
        try {
            NamingService namingService = nacosServiceManager.getNamingService();
            List<Instance> allInstances = namingService.getAllInstances(applicationName);
            List<Instance> availableInstances = allInstances.stream()
                .filter(predicate(instance -> instance.getIp().concat(String.valueOf(instance.getPort())))).toList();
            String syncUrl = "http://".concat(applicationName).concat("/scheduler/task/flush/").concat(operationType);
            availableInstances.forEach(instance -> {
                HttpHeaders headers = new HttpHeaders();
                headers.set(XddHeaderConstant.X_INSTANCE_ROUTE, instance.getInstanceId());
                HttpEntity<TaskDTO> entity = new HttpEntity<>(task, headers);
                ParameterizedTypeReference<CommonResult<String>> responseType = new ParameterizedTypeReference<>() {
                };
                ResponseEntity<CommonResult<String>> exchange =
                    restTemplate.exchange(syncUrl, HttpMethod.POST, entity, responseType);
                logger.info("同步任务结果：{}", exchange.getBody());
            });
        } catch (Exception e) {
            logger.error("同步任务失败", e);
        }
    }


    @SneakyThrows
    private <T> Predicate<T> predicate(Function<T, Object> keyExtractor) {
        String address = InetAddress.getLocalHost().getHostAddress() + serverProperties.getPort();
        return t -> !address.equals(keyExtractor.apply(t));
    }


    /**
     * 更新任务清单
     *
     * @param task
     * @return
     */
    @Override
    public CommonResult<Integer> updateTaskTimer(TaskDTO task) {
        checkValidDate(task);
        CommonResult<Integer> resultDTO = new CommonResult<>();
        resultDTO.setCode(HttpStatus.OK.value());
        if (Objects.isNull(task.getId())) {
            resultDTO.setMsg("schedule task not exists");
            resultDTO.setCode(HttpStatus.NOT_FOUND.value());
            return resultDTO;
        }
        convertDataToExpression(task);
        task.setLastUpdateBy(RequestContext.getCurrentContextUser().getUserId());
        resultDTO.setData(taskTimerDao.updateTaskTimer(task));
        // 清洗任务
        flushScheduleTask(task, OperationType.UPDATE);
        syncPeerNode(task, OperationType.UPDATE.name());
        return resultDTO;
    }

    /**
     * 删除任务记录
     *
     * @param taskId
     * @return
     */
    @Override
    public int deleteTaskTimer(Long taskId) {
        TaskDTO byId = taskTimerDao.findById(taskId);
        if (Objects.isNull(byId)) {
            return 0;
        }
        int data = taskTimerDao.deleteTaskTimer(taskId);
        // 清洗任务
        flushScheduleTask(byId, OperationType.DELETE);
        syncPeerNode(byId, OperationType.DELETE.name());
        return data;
    }

    /**
     * 获取任务列表-分页
     *
     * @param pageRequest
     * @return
     */
    @Override
    public PagedResultListInfo<TaskDTO> getTaskTimer(PageRequest<TaskDTO> pageRequest) {
        PageHelper.startPage(pageRequest.getPageNo(), pageRequest.getPageSize());
        List<TaskDTO> taskTimer = taskTimerDao.getTaskTimer(pageRequest.getEntity());
        userClientService.populateCreateUpdateName(taskTimer);
        return new PagedResultListInfo<>(pageRequest.getPageNo(), pageRequest.getPageSize(), taskTimer);
    }

    /**
     * 获取有效执行任务
     *
     * @return
     */
    @Override
    public List<TaskDTO> getAllEnableTaskTimer() {
        return taskTimerDao.getAllEnableTaskTimer();
    }

    /**
     * 页面数据转换成cron表达式
     *
     * @param task
     */
    private void convertDataToExpression(TaskDTO task) {
        if (!StringUtils.isEmpty(task.getCronInput())) {
            task.setCron(task.getCronInput());
            task.setCronTimeValue(null);
            task.setCronType(null);
            task.setCronValue(null);
            return;
        }
        String cronType = task.getCronType();
        String cronValue = task.getCronValue();
        String cronTimeValue = task.getCronTimeValue();
        List<String> collect = Arrays.stream(cronTimeValue.split(":")).map(Integer::parseInt).map(String::valueOf)
            .collect(Collectors.toList());
        Collections.reverse(collect);
        String timeStr = collect.stream().collect(Collectors.joining(" "));
        StringBuilder cronExpression = new StringBuilder("0 ").append(timeStr);
        CronTypeEum cronTypeEum = CronTypeEum.getCronTypeEum(cronType);
        switch (cronTypeEum) {
            case WEEK:
                cronExpression.append(" * * ").append(cronValue);
                break;
            case MONTH:
                cronExpression.append(" ").append(cronValue).append(" * ?");
                break;
            case DAY:
                cronExpression.append(" ").append(cronValue).append(" * *");
                break;
            case MONTH_LAST:
                cronExpression.append(" L").append(" * ?");
                break;
            case MONTH_WEEK:
                cronExpression.append(" ? * ").append(cronValue.replace(":", "#"));
            default:
                break;
        }
        task.setCron(cronExpression.toString());
    }

    public String getScheduleCode() {
        return SCHEDULE_PREFIX + System.currentTimeMillis();
    }


    public void checkValidDate(TaskDTO taskDTO) {
        LocalDateTime today = LocalDateTime.of(LocalDate.now(), LocalTime.MIN);
        Date validStartDate = taskDTO.getValidStartDate();
        Date validEndDate = taskDTO.getValidEndDate();
        LocalDateTime startLocalDate = LocalDateTime.ofInstant(validStartDate.toInstant(), ZoneId.of("GMT+8"));
        LocalDateTime endLocalDate = LocalDateTime.ofInstant(validEndDate.toInstant(), ZoneId.of("GMT+8"));
        if (startLocalDate.isAfter(endLocalDate) || today.isAfter(endLocalDate)) {
            throw new IllegalArgumentException("任务不在有效期内");
        }
    }

    /**
     * 启用任务
     *
     * @param ids
     */
    public void taskUP(List<Long> ids) {
        List<TaskDTO> taskTimer = taskTimerDao.getTaskListByIds(ids);
        // 更新任务表
        taskTimer.forEach(taskDTO -> {
            checkValidDate(taskDTO);
            ConcurrentHashMap<String, CronTask> extutorMap = SchedulerService.extutorMap;
            if (!extutorMap.containsKey(taskDTO.getScheduleCode())) {
                CronTask cronTask =
                    new CronTask(new SchedulingRunnable(taskDTO, applicationContext, distributedLockService),
                        taskDTO.getCron());
                cronTaskRegistrar.addCronTask(cronTask);
                extutorMap.put(taskDTO.getScheduleCode(), cronTask);
            }
        });
        taskTimerDao.taskUp(ids);
    }

    /**
     * 停止任务
     *
     * @param ids
     */
    @Override
    public void taskDown(List<Long> ids) {
        List<TaskDTO> taskTimer = taskTimerDao.getTaskListByIds(ids);
        taskTimer.forEach(taskDTO -> {
            checkValidDate(taskDTO);
            ConcurrentHashMap<String, CronTask> extutorMap = SchedulerService.extutorMap;
            if (extutorMap.containsKey(taskDTO.getScheduleCode())) {
                CronTask cronTask = extutorMap.get(taskDTO.getScheduleCode());
                cronTaskRegistrar.removeCronTask((SchedulingRunnable) cronTask.getRunnable());
                extutorMap.remove(taskDTO.getScheduleCode());
            }
        });
        taskTimerDao.taskDown(ids);
    }

    /**
     * 手动触发
     *
     * @param ids
     */
    @Override
    public void executeTaskTimer(List<Long> ids) {
        long startTime = System.currentTimeMillis();
        List<TaskDTO> taskTimer = taskTimerDao.getTaskListByIds(ids);
        UserVO currenContexttUser = RequestContext.getCurrentContextUser();
        taskTimer.forEach(taskDTO -> {
            checkValidDate(taskDTO);
            taskDTO.setPerformer(currenContexttUser.getUserName());
            AtomicReference<IScheduleExecutor> atomicReference = new AtomicReference<>();
            switch (ExecuteTypeEnum.getExecuteTypeEnum(taskDTO.getExecuteType())) {
                case Bean ->
                    atomicReference.set(applicationContext.getBean("beanTaskExecutor", IScheduleExecutor.class));
                case URL ->
                    atomicReference.set(applicationContext.getBean("remoteTaskExecutor", IScheduleExecutor.class));
                default -> throw new IllegalArgumentException("不支持的任务类型");
            }
            if (distributedLockService.tryLock(taskDTO.getScheduleCode(), 600000)) {
                threadPoolTaskExecutor.execute(() -> {
                    TaskLogVO taskLogVO = JSON.parseObject(JSON.toJSONString(taskDTO), TaskLogVO.class);
                    IScheduleExecutor iScheduleExecutor = atomicReference.get();
                    try {
                        Runnable runnable = extutorMap.get(taskDTO.getScheduleCode()).getRunnable();
                        iScheduleExecutor.beforeExecute(taskDTO,
                            Optional.of(runnable).orElse(() -> logger.error("未找到任务")));
                        iScheduleExecutor.execute(taskDTO,
                            Optional.of(runnable).orElse(() -> logger.error("未找到任务")));
                        taskLogVO.setCostTime(System.currentTimeMillis() - startTime);
                        taskLogVO.setExecuteTime(new Date(startTime));
                        taskLogVO.setExecuteResult("success");
                        taskLogVO.setResultMessage("success");

                    } catch (Exception e) {
                        taskLogVO.setResultMessage(e.getMessage());
                        taskLogVO.setExecuteResult("fail");
                    } finally {
                        taskLogVO.setExecuteTime(new Date(startTime));
                        taskLogVO.setCostTime(System.currentTimeMillis() - startTime);
                        iScheduleExecutor.afterExecute(taskLogVO);
                        distributedLockService.releaseLock(taskDTO.getScheduleCode());
                    }
                });
            }
        });

    }
}
